package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"log"
	"time"
)

func main() {
	 Pullconsumer()

}
func Pullconsumer()  {
	addr,err := primitive.NewNamesrvAddr("127.0.0.1:9876")
	if err!=nil {
		log.Fatal(err)
	}
	topic:= "Develop"
	//消费者主动拉取消息
	c1,err := rocketmq.NewPullConsumer(
		consumer.WithGroupName("my_service"),
		consumer.WithNameServer(addr),
		consumer.WithRetry(1))
	if err !=nil {
		log.Fatal(err)
	}
	err = c1.Start()
	if err !=nil {
		log.Fatal(err)
	}
	queue:= primitive.MessageQueue{
		Topic  :topic,
		BrokerName : "DESKTOP-VG44KOG",
		QueueId  : 4,
	}
	err = c1.Shutdown()
	if err !=nil {
		fmt.Println("Shutdown Pull Consumer error: ",err)
	}
	offset := int64(0)
	for  {
		//primitive
		resp,err := c1.PullFrom(context.TODO(),queue,offset,10)
		if err!=nil {
			if err == rocketmq.ErrRequestTimeout{
				fmt.Printf("timeout\n")
				time.Sleep(time.Second)
				continue
			}
			fmt.Printf("unexpected error: %v\n",err)
			return
		}

		if  resp.Status == primitive.PullFound{
			//下一个开始偏移
			fmt.Printf("pull message success. nextOffset: %d\n",resp.NextBeginOffset)
			for _,ext:=  range resp.GetMessageExts(){
				fmt.Printf("pull msg: %s\n",ext)
			}
		}
		offset = resp.NextBeginOffset
	}
}


